package com.atbeijing.D05;

import com.atbeijing.D02.SensorReading;
import com.atbeijing.D02.SensorSource;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 *  检测传感器的连续5s温度上升
 *  1,2,3,4,5,1,2
 *
 *  状态变量的使用
 */
public class Example5 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new SensorSource())
                .keyBy(r -> r.id)
                .process( new MyKeyedProcessFunction())
                .print();

        env.execute();
    }

    public static class MyKeyedProcessFunction extends KeyedProcessFunction<String, SensorReading,String>{
        //每个key(每条流)维护一个自己的状态变量,状态变量是单例的,可以存到hdfs

        //状态变量:保存上一个事件的温度
        private ValueState<Double> lastTemp;

        //状态变量:保存报警定时器的时间戳
        private ValueState<Long> timerTs;

        //初始化一次
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            //初始化状态变量,名字和类型
            lastTemp=getRuntimeContext().getState(new ValueStateDescriptor<Double>("last-temp", Types.DOUBLE));
            timerTs=getRuntimeContext().getState(new ValueStateDescriptor<Long>("timer-ts",Types.LONG));
        }

        //每来一条数据调用一次
        @Override
        public void processElement(SensorReading value, Context ctx, Collector<String> out) throws Exception {
            //最近一次温度初始化
            double prevTemp=0.0;
            //当第一条数据来时,lastTemp装态变量为null
            if(lastTemp.value()==null){
                //更新lastTemp
                lastTemp.update(value.temperature);
            }else {
                //不是第一条数据进来,将老的lastTemp保存到prevTemp,新数据存到lastTemp
                prevTemp=lastTemp.value();
                lastTemp.update(value.temperature);
            }

            long ts =0L;
            if (timerTs.value()==null){
                //不做处理
            }else{
                ts=timerTs.value();
            }

            //当第一条数据进来或者进来的数据温度小于上一条数据的温度,证明没有连续升温
            if (prevTemp==0.0 || value.temperature<prevTemp){
                //删除定时器,如果定时器不存在就不执行
                ctx.timerService().deleteProcessingTimeTimer(ts);
                //清空定时器时间戳
                timerTs.clear();
            //如果进来的数据温度大于上一条数据的温度,且定时器的时间戳还没设置,如果设置了说明已经在升温计时了,如果不取消将会报警
            }else if (value.temperature>prevTemp && ts == 0L){
                //设置1s后报警
                ctx.timerService().registerProcessingTimeTimer(ctx.timerService().currentProcessingTime() + 5000L);
                //向状态变量更新定时器时间戳
                timerTs.update(ctx.timerService().currentProcessingTime() + 5000L);
            }

        }

        //定时任务
        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            super.onTimer(timestamp, ctx, out);
            out.collect("连续5s温度上升了！" + "传感器是：" + ctx.getCurrentKey());
            timerTs.clear();
        }
    }
}
